package com.atguigu.flink.process;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeDomain;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Created by 黄凯 on 2023/6/19 0019 9:23
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * TimerService:
 * *     long currentProcessingTime(): 获取当前处理时间(系统时间)
 * *     void registerProcessingTimeTimer(long time)：注册处理时间定时器
 * *     void deleteProcessingTimeTimer(long time):删除处理时间定时器
 * *
 * *     long currentWatermark(): 获取当前水位线(事件时间)
 * *     void registerEventTimeTimer(long time)：注册事件时间定时器
 * *     void deleteEventTimeTimer(long time): 删除事件时间定时器
 * *
 * *
 * *  定时器触发后的处理:
 * *     使用处理函数中的ontimer方法来进行处理。
 * *
 * *  注意: 只有keyBy以后才可以注册定时器。
 * *        Setting timers is only supported on a keyed streams.
 */
public class Flink02_TimerService {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
                        }
                ).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        //注册定时器
        ds.keyBy(
                Event::getUser
        ).process(
                new KeyedProcessFunction<String, Event, String>() {
                    @Override
                    public void processElement(Event value,
                                               KeyedProcessFunction<String, Event, String>.Context ctx, Collector<String> out) throws Exception {

                        //获取定时服务
                        TimerService timerService = ctx.timerService();

                        //处理时间定时器
                        //基于当前处理时间，注册5秒后的定时器
                        //获取当前的处理时间（系统时间）
//                        long processingTime = timerService.currentProcessingTime();

                        //注册定时器
//                        timerService.registerProcessingTimeTimer(processingTime + 5000L);
//
//                        System.out.println("processElement==>注册的处理时间定时器为: " + ctx.getCurrentKey() + "  " + (processingTime + 5000L));


                        //事件时间定时器
                        //基于当前数据的时间，注册5秒后的定时器

                        /**
                         * processElement==>注册的事件时间定时器为: tom  6000
                         * {"ts":1000,"url":"/home","user":"tom"}
                         * processElement==>注册的事件时间定时器为: tom  8000
                         * {"ts":3000,"url":"/home","user":"tom"}
                         * processElement==>注册的事件时间定时器为: tom  11000
                         * {"ts":6000,"url":"/home","user":"tom"}
                         * processElement==>注册的事件时间定时器为: tom  12000
                         * {"ts":7000,"url":"/home","user":"tom"}
                         * onTimer==> 当前触发的定时器的时间为: tom  6000
                         */
                        timerService.registerEventTimeTimer(value.getTs() + 5000L);

                        System.out.println("processElement==>注册的事件时间定时器为: " + ctx.getCurrentKey() + "  "  +  (value.getTs() + 5000L) );

                        out.collect(JSON.toJSONString(value));

                    }

                    /**
                     * 定时器出触发后，要执行的方法
                     * 提前将要做的事情定义到该方法中，等到定时器触发后，会自动执行
                     *
                     * timestamp:触发的定时器的时间
                     *
                     * @param timestamp The timestamp of the firing timer.
                     * @param ctx An {@link OnTimerContext} that allows querying the timestamp, the {@link
                     *     TimeDomain}, and the key of the firing timer and getting a {@link TimerService} for
                     *     registering timers and querying the time. The context is only valid during the invocation
                     *     of this method, do not store it.
                     * @param out The collector for returning result values.
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

                        //处理时间
//                        System.out.println("onTimer==> 当前触发的定时器的时间为: " + ctx.getCurrentKey() + "  " + timestamp);


                        //事件时间
                        System.out.println("onTimer==> 当前触发的定时器的时间为: " + ctx.getCurrentKey() + "  " + timestamp);
                    }
                }
        ).print();

        env.execute();

    }

}
